package com.niodata.dt.fs;

import com.niodata.dt.fs.sftp.SFTPFileSystem;
import java.io.IOException;
import java.net.URI;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;

public class SftpFileSystemBuilder implements FileSystemBuilder {

  @Override
  public FileSystem createFileSystem(String user, Map<String, String> conf) throws IOException {
    Configuration config = new Configuration();
    String host = conf.get(SFTPFileSystem.FS_SFTP_HOST);
    String port = conf.getOrDefault(SFTPFileSystem.FS_SFTP_HOST_PORT, "22");
    config.set("fs.defaultFS", "sftp://" + host + ":" + port);
    config.setClass("fs.ftp.impl", SFTPFileSystem.class, FileSystem.class);
    config.set(SFTPFileSystem.FS_SFTP_HOST, host);
    config.set(SFTPFileSystem.FS_SFTP_HOST_PORT, port);
    String loginUser = conf.getOrDefault("fs.sftp.user", "root");
    config.set(SFTPFileSystem.FS_SFTP_USER_PREFIX + host, loginUser);

    String pwd = conf.get("fs.sftp.password");
    if (pwd != null && pwd.length() > 0) {
      config.set(SFTPFileSystem.FS_SFTP_PASSWORD_PREFIX + host + "." + loginUser, pwd);
    } else {
      config.set(SFTPFileSystem.FS_SFTP_KEYFILE, conf.get(SFTPFileSystem.FS_SFTP_KEYFILE));
    }
    FileSystem fs = new SFTPFileSystem();
    try {
      fs.initialize(new URI(config.get("fs.defaultFS")), config);
    } catch (Exception e) {
      throw new IOException(e);
    }
    return fs;
  }
}
